#
#    Copyright (c) 2025 Project CHIP Authors
#    All rights reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.

# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
# for details about the block below.
#
# === BEGIN CI TEST ARGUMENTS ===
# test-runner-runs:
#   run1:
#     app: ${CAMERA_APP}
#     app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
#     script-args: >
#       --storage-path admin_storage.json
#       --commissioning-method on-network
#       --discriminator 1234
#       --passcode 20202021
#       --PICS src/app/tests/suites/certification/ci-pics-values
#       --trace-to json:${TRACE_TEST_JSON}.json
#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
#       --endpoint 1
#     factory-reset: true
#     quiet: true
# === END CI TEST ARGUMENTS ===

import logging

from mobly import asserts
from TC_AVSMTestBase import AVSMTestBase, wmark_osd_matcher

import matter.clusters as Clusters
from matter.interaction_model import InteractionModelError, Status
from matter.testing.event_attribute_reporting import AttributeSubscriptionHandler
from matter.testing.matter_testing import MatterBaseTest, TestStep, default_matter_test_main, has_feature, run_if_endpoint_matches

log = logging.getLogger(__name__)


class TC_AVSM_2_8(MatterBaseTest, AVSMTestBase):
    def desc_TC_AVSM_2_8(self) -> str:
        return "[TC-AVSM-2.8] Validate Video Stream Modification functionality with Server as DUT"

    def pics_TC_AVSM_2_8(self):
        return ["AVSM.S"]

    def steps_TC_AVSM_2_8(self) -> list[TestStep]:
        return [
            TestStep("precondition", "DUT commissioned and preconditions", is_commissioning=True),
            TestStep(
                1,
                "TH reads FeatureMap attribute from CameraAVStreamManagement Cluster on DUT",
                "Verify VDO & (WMARK|OSD) is supported.",
            ),
            TestStep(
                2,
                "TH reads AllocatedVideoStreams attribute from CameraAVStreamManagement Cluster on DUT",
                "Verify the number of allocated video streams in the list is 1. Store StreamID as aStreamID. If WMARK is supported, store WaterMarkEnabled as aWmark. If OSD is supported, store OSDEnabled as aOSD.",
            ),
            TestStep(
                3,
                "TH sends the VideoStreamModify command with VideoStreamID set to aStreamID. No WaterMarkEnabled or OSDEnabled provided.",
                "DUT responds with an INVALID_COMMAND status code.",
            ),
            TestStep(
                4,
                "TH sends the VideoStreamModify command with VideoStreamID set to aStreamID + 1.",
                "DUT responds with an NOT_FOUND status code.",
            ),
            TestStep(
                5,
                "Establish a subscription to the AllocatedVideoStreams attribute"
            ),
            TestStep(
                6,
                "TH sends the VideoStreamModify command with VideoStreamID set to aStreamID. If WMARK is supported, set WaterMarkEnabled to !aWmark and if OSD is supported, set OSDEnabled to !aOSD in the command.",
                "DUT responds with a SUCCESS status code.",
            ),
            TestStep(
                7,
                "Wait until a subscription report with AllocatedVideoStreams attribute is received",
            ),
            TestStep(
                8,
                "TH reads AllocatedVideoStreams attribute from CameraAVStreamManagement Cluster on DUT",
                "Verify the following: If WMARK is supported, verify WaterMarkEnabled == !aWmark. If OSD is supported, verify OSDEnabled == !aOSD.",
            ),
        ]

    @run_if_endpoint_matches(
        has_feature(Clusters.CameraAvStreamManagement, Clusters.CameraAvStreamManagement.Bitmaps.Feature.kVideo)
        and (
            has_feature(Clusters.CameraAvStreamManagement, Clusters.CameraAvStreamManagement.Bitmaps.Feature.kWatermark)
            or has_feature(Clusters.CameraAvStreamManagement, Clusters.CameraAvStreamManagement.Bitmaps.Feature.kOnScreenDisplay)
        )
    )
    async def test_TC_AVSM_2_8(self):
        endpoint = self.get_endpoint()
        cluster = Clusters.CameraAvStreamManagement
        attr = Clusters.CameraAvStreamManagement.Attributes
        commands = Clusters.CameraAvStreamManagement.Commands

        self.step("precondition")
        # Commission DUT - already done
        await self.precondition_one_allocated_video_stream()

        self.step(1)
        aFeatureMap = await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attr.FeatureMap)
        log.info(f"Rx'd FeatureMap: {aFeatureMap}")
        vdoSupport = (aFeatureMap & cluster.Bitmaps.Feature.kVideo) > 0
        wmarkSupport = (aFeatureMap & cluster.Bitmaps.Feature.kWatermark) > 0
        osdSupport = (aFeatureMap & cluster.Bitmaps.Feature.kOnScreenDisplay) > 0
        asserts.assert_true((vdoSupport and (wmarkSupport or osdSupport)), "VDO & (WMARK|OSD) is not supported.")

        self.step(2)
        aAllocatedVideoStreams = await self.read_single_attribute_check_success(
            endpoint=endpoint, cluster=cluster, attribute=attr.AllocatedVideoStreams
        )
        log.info(f"Rx'd AllocatedVideoStreams: {aAllocatedVideoStreams}")
        asserts.assert_equal(len(aAllocatedVideoStreams), 1, "The number of allocated video streams in the list is not 1")
        aStreamID = aAllocatedVideoStreams[0].videoStreamID
        aWmark = aAllocatedVideoStreams[0].watermarkEnabled
        aOSD = aAllocatedVideoStreams[0].OSDEnabled

        self.step(3)
        try:
            videoStreamModifyCmd = commands.VideoStreamModify(
                videoStreamID=aStreamID,
                watermarkEnabled=None,
                OSDEnabled=None,
            )
            await self.send_single_cmd(endpoint=endpoint, cmd=videoStreamModifyCmd)
            asserts.fail("Unexpected success when expecting INVALID_COMMAND due to absence of WatermarkEnabled and OSDEnabled")
        except InteractionModelError as e:
            asserts.assert_equal(e.status, Status.InvalidCommand, "Unexpected error when expecting INVALID_COMMAND")
            pass

        self.step(4)
        try:
            videoStreamModifyCmd = commands.VideoStreamModify(
                videoStreamID=aStreamID + 1,
                watermarkEnabled=None if aWmark is None else not aWmark,
                OSDEnabled=None if aOSD is None else not aOSD,
            )
            await self.send_single_cmd(endpoint=endpoint, cmd=videoStreamModifyCmd)
            asserts.fail("Unexpected success when expecting NOT_FOUND due to wrong streamID")
        except InteractionModelError as e:
            asserts.assert_equal(e.status, Status.NotFound, "Unexpected error when expecting NOT_FOUND")
            pass

        self.step(5)
        # Establish subscription to AllocatedVideoStreams
        sub_handler = AttributeSubscriptionHandler(cluster, attr.AllocatedVideoStreams)
        await sub_handler.start(self.default_controller, self.dut_node_id, endpoint=endpoint, min_interval_sec=0, max_interval_sec=30, keepSubscriptions=False)

        sub_handler.reset()

        self.step(6)
        try:
            videoStreamModifyCmd = commands.VideoStreamModify(
                videoStreamID=aStreamID,
                watermarkEnabled=None if aWmark is None else not aWmark,
                OSDEnabled=None if aOSD is None else not aOSD,
            )
            await self.send_single_cmd(endpoint=endpoint, cmd=videoStreamModifyCmd)
        except InteractionModelError as e:
            asserts.assert_equal(e.status, Status.Success, "Unexpected error returned")
            pass

        self.step(7)
        expected_wmark = None if aWmark is None else not aWmark
        expected_osd = None if aOSD is None else not aOSD
        sub_handler.await_all_expected_report_matches(expected_matchers=[wmark_osd_matcher(
            attr.AllocatedVideoStreams, expected_wmark, expected_osd, wmarkSupport, osdSupport)], timeout_sec=20)

        self.step(8)
        aAllocatedVideoStreams = await self.read_single_attribute_check_success(
            endpoint=endpoint, cluster=cluster, attribute=attr.AllocatedVideoStreams
        )
        log.info(f"Rx'd AllocatedVideoStreams: {aAllocatedVideoStreams}")
        if wmarkSupport:
            asserts.assert_equal(aAllocatedVideoStreams[0].watermarkEnabled, not aWmark, "WaterMarkEnabled is not !aWmark")
        if osdSupport:
            asserts.assert_equal(aAllocatedVideoStreams[0].OSDEnabled, not aOSD, "OSDEnabled is not !aOSD")

        # Clear all allocated streams
        aAllocatedVideoStreams = await self.read_single_attribute_check_success(
            endpoint=endpoint, cluster=cluster, attribute=attr.AllocatedVideoStreams
        )

        for stream in aAllocatedVideoStreams:
            try:
                await self.send_single_cmd(endpoint=endpoint, cmd=commands.VideoStreamDeallocate(videoStreamID=(stream.videoStreamID)))
            except InteractionModelError as e:
                asserts.assert_equal(e.status, Status.Success, "Unexpected error returned")


if __name__ == "__main__":
    default_matter_test_main()
